package com.shujia.spark.core

import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.Logger

object Demo27PageRank extends Logging {

  val logger: Logger = log

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo27PageRank")

    val sc: SparkContext = new SparkContext(conf)

    val pageLineRDD: RDD[String] = sc.textFile("spark/data/Page.txt")

    val pageNum: Long = pageLineRDD.count()

    val pageRDD: RDD[(String, List[String])] = pageLineRDD.map(line => {
      val splits: Array[String] = line.split("->")
      val page: String = splits(0)
      (page, splits(1).split(",").toList)
    })

    val initPR: Double = 1.0

    // 给每个页面附上初值
    var pagePRRDD: RDD[(String, List[String], Double)] = pageRDD.map(t2 => (t2._1, t2._2, initPR))

    var cnt: Int = 0
    val maxIterCnt: Int = 20 // 最大迭代次数

    while (cnt < maxIterCnt) {
      cnt += 1

      // 将PR值平均分给每一个指向的页面
      val assignPRRDD: RDD[(String, Double)] = pagePRRDD.flatMap(t3 => {
        val avgPR: Double = t3._3 / t3._2.size
        t3._2.map(p => (p, avgPR))
      })

      // 统计每个页面新的PR值
      val pageNewPRRDD: RDD[(String, Double)] = assignPRRDD.reduceByKey(_ + _)
      pageNewPRRDD.foreach(println)
      // 计算平均误差，新的PR值减去上一次每个页面的PR值，再除以页面的数量

      val joinRDD: RDD[(String, ((List[String], Double), Double))] = pagePRRDD.map(t3 => (t3._1, (t3._2, t3._3))).join(pageNewPRRDD)

      val avgDiff: Double = joinRDD.map {
        case (page: String, ((pages: List[String], oldPR: Double), newPR: Double)) =>
          Math.abs(oldPR - newPR)
      }.reduce(_ + _) / pageNum

      logger.info(s"当前已经迭代了${cnt}次，平均误差为：$avgDiff")

      if (avgDiff < 0.0001) {
        // 如果平均误差小于该阈值，则跳出循环迭代
        logger.debug("平均误差小于阈值，迭代停止")
        cnt = maxIterCnt + 1
      }

      val newPagePRRDD: RDD[(String, List[String], Double)] = joinRDD.map {
        case (page: String, ((pages: List[String], oldPR: Double), newPR: Double)) =>
          (page, pages, newPR)
      }

      pagePRRDD = newPagePRRDD


    }
    if (cnt == maxIterCnt) {
      logger.debug("达到了最大的迭代次数")
    }

  }

}
